package com.iot.core.base;

import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.profile.DefaultProfile;
import com.iot.core.receiver.BaseReceiver;
import com.iot.core.sender.IotDefaultSender;
import com.iot.core.sender.IotSender;
import lombok.extern.slf4j.Slf4j;
import org.apache.qpid.jms.JmsConnectionListener;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import java.net.URI;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@EnableAsync
@Configuration
@EnableConfigurationProperties(value = {IotConnectParam.class})
public class IotConfig {

    @Bean
    public Executor iotExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(10);
        executor.setKeepAliveSeconds(60);
        executor.setQueueCapacity(500);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        executor.initialize();
        return executor;
    }

    @Bean
    public IAcsClient iAcsClient(IotConnectParam iotConnectParam) {
        DefaultProfile profile = DefaultProfile.getProfile(iotConnectParam.getRegionId(),
                iotConnectParam.getAccessKey(), iotConnectParam.getAccessSecret());
        return new DefaultAcsClient(profile);
    }

    @Bean
    public JmsConnectionListener jmsConnectionListener() {
        return new JmsConnectionListener() {
            @Override
            public void onConnectionEstablished(URI remoteURI) {
                log.info("连接成功, remoteUri:{}", remoteURI);
            }

            @Override
            public void onConnectionFailure(Throwable error) {
                log.error("连接失败, {}", error.getMessage());
            }

            @Override
            public void onConnectionInterrupted(URI remoteURI) {
                log.info("连接中断, remoteUri:{}", remoteURI);
            }

            @Override
            public void onConnectionRestored(URI remoteURI) {
                log.info("连接恢复, remoteUri:{}", remoteURI);
            }

            @Override
            public void onInboundMessage(JmsInboundMessageDispatch envelope) {
            }

            @Override
            public void onSessionClosed(Session session, Throwable cause) {
            }

            @Override
            public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {
            }

            @Override
            public void onProducerClosed(MessageProducer producer, Throwable cause) {
            }
        };
    }

    @Bean
    public MessageListener messageListener(List<BaseReceiver> baseReceiverList) {
        return message -> {
            //接受消息，发给不同的订阅器
            IotMessage iotMessage = new IotMessage(message);
            baseReceiverList.forEach(receiver -> {
                if (MqttTopic.isMatched(receiver.getSubTopic().getTopicFilter(), iotMessage.getTopic())) {
                    receiver.receive(iotMessage);
                }
            });
        };
    }

    @Bean(initMethod = "connect", destroyMethod = "disconnect")
    public IotConnection iotConnection(JmsConnectionListener jmsConnectionListener,
                                       MessageListener messageListener,
                                       IotConnectParam iotConnectParam) {
        IotEnvironment iotEnvironment = IotEnvironment.builder()
                .iotConnectParam(iotConnectParam)
                .connectCount(3)
                .signMethod(IotConstant.SIGN_METHOD_HMACSHA1)
                .messageListener(messageListener)
                .connectionListener(jmsConnectionListener)
                .build();
        return new IotDefaultIConnection(iotEnvironment);
    }

    @Bean
    public IotSender mqttSender(IotConnectParam iotConnectParam, IAcsClient iAcsClient) {
        return new IotDefaultSender(iAcsClient, iotConnectParam);
    }

}
